/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.sdk.coders;



import com.bff.gaia.unified.sdk.annotations.Internal;

import com.bff.gaia.unified.sdk.io.FileIO;

import com.bff.gaia.unified.sdk.io.ReadableFileCoder;

import com.bff.gaia.unified.sdk.io.fs.MetadataCoder;

import com.bff.gaia.unified.sdk.io.fs.ResourceId;

import com.bff.gaia.unified.sdk.io.fs.ResourceIdCoder;

import com.bff.gaia.unified.sdk.transforms.SerializableFunction;

import com.bff.gaia.unified.sdk.transforms.windowing.IntervalWindow;

import com.bff.gaia.unified.sdk.util.CoderUtils;

import com.bff.gaia.unified.sdk.util.common.ReflectHelpers;

import com.bff.gaia.unified.sdk.values.KV;

import com.bff.gaia.unified.sdk.values.TimestampedValue;

import com.bff.gaia.unified.sdk.values.TypeDescriptor;

import com.bff.gaia.unified.vendor.guava.com.google.common.annotations.VisibleForTesting;

import com.bff.gaia.unified.vendor.guava.com.google.common.collect.*;

import com.bff.gaia.unified.sdk.io.fs.MatchResult;

import org.joda.time.Instant;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;



import javax.annotation.Nullable;

import java.lang.reflect.ParameterizedType;

import java.lang.reflect.Type;

import java.lang.reflect.TypeVariable;

import java.lang.reflect.WildcardType;

import java.util.*;



import static com.bff.gaia.unified.vendor.guava.com.google.common.base.Preconditions.checkArgument;



/**

 * A {@link CoderRegistry} allows creating a {@link Coder} for a given Java {@link Class class} or

 * {@link TypeDescriptor type descriptor}.

 *

 * <p>Creation of the {@link Coder} is delegated to one of the many registered {@link CoderProvider

 * coder providers} based upon the registration order.

 *

 * <p>By default, the {@link CoderProvider coder provider} precedence order is as follows:

 *

 * <ul>

 *   <li>Coder providers registered programmatically with {@link

 *       CoderRegistry#registerCoderProvider(CoderProvider)}.

 *   <li>A default coder provider for common Java (Byte, Double, List, ...) and Apache Unified (KV,

 *       ...) types.

 *   <li>Coder providers registered automatically through a {@link CoderProviderRegistrar} using a

 *       {@link ServiceLoader}. Note that the {@link ServiceLoader} registration order is consistent

 *       but may change due to the addition or removal of libraries exposed to the application. This

 *       can impact the coder returned if multiple coder providers are capable of supplying a coder

 *       for the specified type.

 * </ul>

 *

 * <p>Note that if multiple {@link CoderProvider coder providers} can provide a {@link Coder} for a

 * given type, the precedence order above defines which {@link CoderProvider} is chosen.

 */

public class CoderRegistry {



  private static final Logger LOG = LoggerFactory.getLogger(CoderRegistry.class);

  private static final List<CoderProvider> REGISTERED_CODER_FACTORIES;



  /** A {@link CoderProvider} for common Java SDK and Apache Unified SDK types. */

  private static class CommonTypes extends CoderProvider {

    private final Map<Class<?>, CoderProvider> commonTypesToCoderProviders;



    private CommonTypes() {

      ImmutableMap.Builder<Class<?>, CoderProvider> builder = ImmutableMap.builder();

      builder.put(

          Boolean.class, CoderProviders.fromStaticMethods(Boolean.class, BooleanCoder.class));

      builder.put(Byte.class, CoderProviders.fromStaticMethods(Byte.class, ByteCoder.class));

      builder.put(BitSet.class, CoderProviders.fromStaticMethods(BitSet.class, BitSetCoder.class));

      builder.put(

          FloatCoder.class, CoderProviders.fromStaticMethods(Float.class, FloatCoder.class));

      builder.put(Double.class, CoderProviders.fromStaticMethods(Double.class, DoubleCoder.class));

      builder.put(

          Instant.class, CoderProviders.fromStaticMethods(Instant.class, InstantCoder.class));

      builder.put(

          Integer.class, CoderProviders.fromStaticMethods(Integer.class, VarIntCoder.class));

      builder.put(

          Iterable.class, CoderProviders.fromStaticMethods(Iterable.class, IterableCoder.class));

      builder.put(KV.class, CoderProviders.fromStaticMethods(KV.class, KvCoder.class));

      builder.put(List.class, CoderProviders.fromStaticMethods(List.class, ListCoder.class));

      builder.put(Long.class, CoderProviders.fromStaticMethods(Long.class, VarLongCoder.class));

      builder.put(Map.class, CoderProviders.fromStaticMethods(Map.class, MapCoder.class));

      builder.put(

          MatchResult.Metadata.class, CoderProviders.fromStaticMethods(MatchResult.Metadata.class, MetadataCoder.class));

      builder.put(

          ResourceId.class,

          CoderProviders.fromStaticMethods(ResourceId.class, ResourceIdCoder.class));

      builder.put(

          FileIO.ReadableFile.class,

          CoderProviders.fromStaticMethods(FileIO.ReadableFile.class, ReadableFileCoder.class));

      builder.put(Set.class, CoderProviders.fromStaticMethods(Set.class, SetCoder.class));

      builder.put(

          String.class, CoderProviders.fromStaticMethods(String.class, StringUtf8Coder.class));

      builder.put(

          TimestampedValue.class,

          CoderProviders.fromStaticMethods(

              TimestampedValue.class, TimestampedValue.TimestampedValueCoder.class));

      builder.put(Void.class, CoderProviders.fromStaticMethods(Void.class, VoidCoder.class));

      builder.put(

          byte[].class, CoderProviders.fromStaticMethods(byte[].class, ByteArrayCoder.class));

      builder.put(

          IntervalWindow.class,

          CoderProviders.forCoder(

              TypeDescriptor.of(IntervalWindow.class), IntervalWindow.getCoder()));

      commonTypesToCoderProviders = builder.build();

    }



    @Override

    public <T> Coder<T> coderFor(

		TypeDescriptor<T> typeDescriptor, List<? extends Coder<?>> componentCoders)

        throws CannotProvideCoderException {

      CoderProvider factory = commonTypesToCoderProviders.get(typeDescriptor.getRawType());

      if (factory == null) {

        throw new CannotProvideCoderException(

            String.format("%s is not one of the common types.", typeDescriptor));

      }

      return factory.coderFor(typeDescriptor, componentCoders);

    }

  }



  static {

    // Register the standard coders first so they are chosen over ServiceLoader ones

    List<CoderProvider> codersToRegister = new ArrayList<>();

    codersToRegister.add(new CommonTypes());



    // Enumerate all the CoderRegistrars in a deterministic order, adding all coders to register

    Set<CoderProviderRegistrar> registrars = Sets.newTreeSet(ReflectHelpers.ObjectsClassComparator.INSTANCE);

    registrars.addAll(

        Lists.newArrayList(

            ServiceLoader.load(CoderProviderRegistrar.class, ReflectHelpers.findClassLoader())));



    // DefaultCoder should have the highest precedence and SerializableCoder the lowest

    codersToRegister.addAll(new DefaultCoder.DefaultCoderProviderRegistrar().getCoderProviders());

    for (CoderProviderRegistrar registrar : registrars) {

      codersToRegister.addAll(registrar.getCoderProviders());

    }

    codersToRegister.add(SerializableCoder.getCoderProvider());



    REGISTERED_CODER_FACTORIES = ImmutableList.copyOf(codersToRegister);

  }



  /**

   * Creates a CoderRegistry containing registrations for all standard coders part of the core Java

   * Apache Unified SDK and also any registrations provided by {@link CoderProviderRegistrar coder

   * registrars}.

   *

   * <p>Multiple registrations which can produce a coder for a given type result in a Coder created

   * by the (in order of precedence):

   *

   * <ul>

   *   <li>{@link CoderProvider coder providers} registered programmatically through {@link

   *       CoderRegistry#registerCoderProvider}.

   *   <li>{@link CoderProvider coder providers} for core types found within the Apache Unified Java

   *       SDK being used.

   *   <li>The {@link CoderProvider coder providers} from the {@link CoderProviderRegistrar} with

   *       the lexicographically smallest {@link Class#getName() class name} being used.

   * </ul>

   */

  public static CoderRegistry createDefault() {

    return new CoderRegistry();

  }



  private CoderRegistry() {

    coderProviders = new ArrayDeque<>(REGISTERED_CODER_FACTORIES);

  }



  /**

   * Registers {@code coderProvider} as a potential {@link CoderProvider} which can produce {@code

   * Coder} instances.

   *

   * <p>This method prioritizes this {@link CoderProvider} over all prior registered coders.

   *

   * <p>See {@link CoderProviders} for common {@link CoderProvider} patterns.

   */

  public void registerCoderProvider(CoderProvider coderProvider) {

    coderProviders.addFirst(coderProvider);

  }



  /**

   * Registers the provided {@link Coder} for the given class.

   *

   * <p>Note that this is equivalent to {@code registerCoderForType(TypeDescriptor.of(clazz))}. See

   * {@link #registerCoderForType(TypeDescriptor, Coder)} for further details.

   */

  public void registerCoderForClass(Class<?> clazz, Coder<?> coder) {

    registerCoderForType(TypeDescriptor.of(clazz), coder);

  }



  /**

   * Registers the provided {@link Coder} for the given type.

   *

   * <p>Note that this is equivalent to {@code registerCoderProvider(CoderProviders.forCoder(type,

   * coder))}. See {@link #registerCoderProvider} and {@link CoderProviders#forCoder} for further

   * details.

   */

  public void registerCoderForType(TypeDescriptor<?> type, Coder<?> coder) {

    registerCoderProvider(CoderProviders.forCoder(type, coder));

  }



  /**

   * Returns the {@link Coder} to use for values of the given class.

   *

   * @throws CannotProvideCoderException if a {@link Coder} cannot be provided

   */

  public <T> Coder<T> getCoder(Class<T> clazz) throws CannotProvideCoderException {

    return getCoder(TypeDescriptor.of(clazz));

  }



  /**

   * Returns the {@link Coder} to use for values of the given type.

   *

   * @throws CannotProvideCoderException if a {@link Coder} cannot be provided

   */

  public <T> Coder<T> getCoder(TypeDescriptor<T> type) throws CannotProvideCoderException {

    return getCoderFromTypeDescriptor(type, HashMultimap.create());

  }



  /**

   * Returns the {@link Coder} for values of the given type, where the given input type uses the

   * given {@link Coder}.

   *

   * @throws CannotProvideCoderException if a {@link Coder} cannot be provided

   * @deprecated This method is to change in an unknown backwards incompatible way once support for

   *     this functionality is refined.

   */

  @Deprecated

  @Internal

  public <InputT, OutputT> Coder<OutputT> getCoder(

      TypeDescriptor<OutputT> typeDescriptor,

      TypeDescriptor<InputT> inputTypeDescriptor,

      Coder<InputT> inputCoder)

      throws CannotProvideCoderException {

    checkArgument(typeDescriptor != null);

    checkArgument(inputTypeDescriptor != null);

    checkArgument(inputCoder != null);

    return getCoderFromTypeDescriptor(

        typeDescriptor, getTypeToCoderBindings(inputTypeDescriptor.getType(), inputCoder));

  }



  /**

   * Returns the {@link Coder} to use on elements produced by this function, given the {@link Coder}

   * used for its input elements.

   *

   * @throws CannotProvideCoderException if a {@link Coder} cannot be provided

   * @deprecated This method is to change in an unknown backwards incompatible way once support for

   *     this functionality is refined.

   */

  @Deprecated

  @Internal

  public <InputT, OutputT> Coder<OutputT> getOutputCoder(

	  SerializableFunction<InputT, OutputT> fn, Coder<InputT> inputCoder)

      throws CannotProvideCoderException {



    ParameterizedType fnType =

        (ParameterizedType)

            TypeDescriptor.of(fn.getClass()).getSupertype(SerializableFunction.class).getType();



    return getCoder(

        fn.getClass(),

        SerializableFunction.class,

        ImmutableMap.of(fnType.getActualTypeArguments()[0], inputCoder),

        SerializableFunction.class.getTypeParameters()[1]);

  }



  /**

   * Returns the {@link Coder} to use for the specified type parameter specialization of the

   * subclass, given {@link Coder Coders} to use for all other type parameters (if any).

   *

   * @throws CannotProvideCoderException if a {@link Coder} cannot be provided

   * @deprecated This method is to change in an unknown backwards incompatible way once support for

   *     this functionality is refined.

   */

  @Deprecated

  @Internal

  public <T, OutputT> Coder<OutputT> getCoder(

      Class<? extends T> subClass,

      Class<T> baseClass,

      Map<Type, ? extends Coder<?>> knownCoders,

      TypeVariable<?> param)

      throws CannotProvideCoderException {



    Map<Type, Coder<?>> inferredCoders = getDefaultCoders(subClass, baseClass, knownCoders);



    @SuppressWarnings("unchecked")

	Coder<OutputT> paramCoderOrNull = (Coder<OutputT>) inferredCoders.get(param);

    if (paramCoderOrNull != null) {

      return paramCoderOrNull;

    } else {

      throw new CannotProvideCoderException("Cannot infer coder for type parameter " + param);

    }

  }



  /////////////////////////////////////////////////////////////////////////////



  /**

   * Returns a {@code Map} from each of {@code baseClass}'s type parameters to the {@link Coder} to

   * use for it, in the context of {@code subClass}'s specialization of {@code baseClass}.

   *

   * <p>If no {@link Coder} can be inferred for a particular type parameter, then that type variable

   * will be absent from the returned {@code Map}.

   *

   * <p>For example, if {@code baseClass} is {@code Map.class}, where {@code Map<K, V>} has type

   * parameters {@code K} and {@code V}, and {@code subClass} extends {@code Map<String, Integer>}

   * then the result will map the type variable {@code K} to a {@code Coder<String>} and the type

   * variable {@code V} to a {@code Coder<Integer>}.

   *

   * <p>The {@code knownCoders} parameter can be used to provide known {@link Coder Coders} for any

   * of the parameters; these will be used to infer the others.

   *

   * <p>Note that inference is attempted for every type variable. For a type {@code MyType<One, Two,

   * Three>} inference will be attempted for all of {@code One}, {@code Two}, {@code Three}, even if

   * the requester only wants a {@link Coder} for {@code Two}.

   *

   * <p>For this reason {@code getDefaultCoders} (plural) does not throw an exception if a {@link

   * Coder} for a particular type variable cannot be inferred, but merely omits the entry from the

   * returned {@code Map}. It is the responsibility of the caller (usually {@link

   * #getCoderFromTypeDescriptor} to extract the desired coder or throw a {@link

   * CannotProvideCoderException} when appropriate.

   *

   * @param subClass the concrete type whose specializations are being inferred

   * @param baseClass the base type, a parameterized class

   * @param knownCoders a map corresponding to the set of known {@link Coder Coders} indexed by

   *     parameter name

   */

  private <T> Map<Type, Coder<?>> getDefaultCoders(

      Class<? extends T> subClass, Class<T> baseClass, Map<Type, ? extends Coder<?>> knownCoders) {

    TypeVariable<Class<T>>[] typeParams = baseClass.getTypeParameters();

    Coder<?>[] knownCodersArray = new Coder<?>[typeParams.length];

    for (int i = 0; i < typeParams.length; i++) {

      knownCodersArray[i] = knownCoders.get(typeParams[i]);

    }

    Coder<?>[] resultArray = getDefaultCoders(subClass, baseClass, knownCodersArray);

    Map<Type, Coder<?>> result = new HashMap<>();

    for (int i = 0; i < typeParams.length; i++) {

      if (resultArray[i] != null) {

        result.put(typeParams[i], resultArray[i]);

      }

    }

    return result;

  }



  /**

   * Returns an array listing, for each of {@code baseClass}'s type parameters, the {@link Coder} to

   * use for it, in the context of {@code subClass}'s specialization of {@code baseClass}.

   *

   * <p>If a {@link Coder} cannot be inferred for a type variable, its slot in the resulting array

   * will be {@code null}.

   *

   * <p>For example, if {@code baseClass} is {@code Map.class}, where {@code Map<K, V>} has type

   * parameters {@code K} and {@code V} in that order, and {@code subClass} extends {@code

   * Map<String, Integer>} then the result will contain a {@code Coder<String>} and a {@code

   * Coder<Integer>}, in that order.

   *

   * <p>The {@code knownCoders} parameter can be used to provide known {@link Coder Coders} for any

   * of the type parameters. These will be used to infer the others. If non-null, the length of this

   * array must match the number of type parameters of {@code baseClass}, and simply be filled with

   * {@code null} values for each type parameters without a known {@link Coder}.

   *

   * <p>Note that inference is attempted for every type variable. For a type {@code MyType<One, Two,

   * Three>} inference will will be attempted for all of {@code One}, {@code Two}, {@code Three},

   * even if the requester only wants a {@link Coder} for {@code Two}.

   *

   * <p>For this reason {@code getDefaultCoders} (plural) does not throw an exception if a {@link

   * Coder} for a particular type variable cannot be inferred. Instead, it results in a {@code null}

   * in the array. It is the responsibility of the caller (usually {@link

   * #getCoderFromTypeDescriptor} to extract the desired coder or throw a {@link

   * CannotProvideCoderException} when appropriate.

   *

   * @param subClass the concrete type whose specializations are being inferred

   * @param baseClass the base type, a parameterized class

   * @param knownCoders an array corresponding to the set of base class type parameters. Each entry

   *     can be either a {@link Coder} (in which case it will be used for inference) or {@code null}

   *     (in which case it will be inferred). May be {@code null} to indicate the entire set of

   *     parameters should be inferred.

   * @throws IllegalArgumentException if baseClass doesn't have type parameters or if the length of

   *     {@code knownCoders} is not equal to the number of type parameters of {@code baseClass}.

   */

  private <T> Coder<?>[] getDefaultCoders(

      Class<? extends T> subClass, Class<T> baseClass, @Nullable Coder<?>[] knownCoders) {

    Type type = TypeDescriptor.of(subClass).getSupertype(baseClass).getType();

    if (!(type instanceof ParameterizedType)) {

      throw new IllegalArgumentException(type + " is not a ParameterizedType");

    }

    ParameterizedType parameterizedType = (ParameterizedType) type;

    Type[] typeArgs = parameterizedType.getActualTypeArguments();

    if (knownCoders == null) {

      knownCoders = new Coder<?>[typeArgs.length];

    } else if (typeArgs.length != knownCoders.length) {

      throw new IllegalArgumentException(

          String.format(

              "Class %s has %d parameters, but %d coders are requested.",

              baseClass.getCanonicalName(), typeArgs.length, knownCoders.length));

    }



    SetMultimap<Type, Coder<?>> context = HashMultimap.create();

    for (int i = 0; i < knownCoders.length; i++) {

      if (knownCoders[i] != null) {

        try {

          verifyCompatible(knownCoders[i], typeArgs[i]);

        } catch (IncompatibleCoderException exn) {

          throw new IllegalArgumentException(

              String.format(

                  "Provided coders for type arguments of %s contain incompatibilities:"

                      + " Cannot encode elements of type %s with coder %s",

                  baseClass, typeArgs[i], knownCoders[i]),

              exn);

        }

        context.putAll(getTypeToCoderBindings(typeArgs[i], knownCoders[i]));

      }

    }



    Coder<?>[] result = new Coder<?>[typeArgs.length];

    for (int i = 0; i < knownCoders.length; i++) {

      if (knownCoders[i] != null) {

        result[i] = knownCoders[i];

      } else {

        try {

          result[i] = getCoderFromTypeDescriptor(TypeDescriptor.of(typeArgs[i]), context);

        } catch (CannotProvideCoderException exc) {

          result[i] = null;

        }

      }

    }

    return result;

  }



  /**

   * Thrown when a {@link Coder} cannot possibly encode a type, yet has been proposed as a {@link

   * Coder} for that type.

   */

  @VisibleForTesting

  static class IncompatibleCoderException extends RuntimeException {

    private Coder<?> coder;

    private Type type;



    IncompatibleCoderException(String message, Coder<?> coder, Type type) {

      super(message);

      this.coder = coder;

      this.type = type;

    }



    IncompatibleCoderException(String message, Coder<?> coder, Type type, Throwable cause) {

      super(message, cause);

      this.coder = coder;

      this.type = type;

    }



    public Coder<?> getCoder() {

      return coder;

    }



    public Type getType() {

      return type;

    }

  }



  /**

   * Returns {@code true} if the given {@link Coder} can possibly encode elements of the given type.

   */

  @VisibleForTesting

  static <T, CoderT extends Coder<T>, CandidateT> void verifyCompatible(

      CoderT coder, Type candidateType) throws IncompatibleCoderException {



    // Various representations of the coder's class

    @SuppressWarnings("unchecked")

    Class<CoderT> coderClass = (Class<CoderT>) coder.getClass();

    TypeDescriptor<CoderT> coderDescriptor = TypeDescriptor.of(coderClass);



    // Various representations of the actual coded type

    @SuppressWarnings("unchecked")

	TypeDescriptor<T> codedDescriptor = CoderUtils.getCodedType(coderDescriptor);

    @SuppressWarnings("unchecked")

    Class<T> codedClass = (Class<T>) codedDescriptor.getRawType();

    Type codedType = codedDescriptor.getType();



    // Various representations of the candidate type

    @SuppressWarnings("unchecked")

	TypeDescriptor<CandidateT> candidateDescriptor =

        (TypeDescriptor<CandidateT>) TypeDescriptor.of(candidateType);

    @SuppressWarnings("unchecked")

    Class<CandidateT> candidateClass = (Class<CandidateT>) candidateDescriptor.getRawType();



    // If coder has type Coder<T> where the actual value of T is lost

    // to erasure, then we cannot rule it out.

    if (candidateType instanceof TypeVariable) {

      return;

    }



    // If the raw types are not compatible, we can certainly rule out

    // coder compatibility

    if (!codedClass.isAssignableFrom(candidateClass)) {

      throw new IncompatibleCoderException(

          String.format(

              "Cannot encode elements of type %s with coder %s because the"

                  + " coded type %s is not assignable from %s",

              candidateType, coder, codedClass, candidateType),

          coder,

          candidateType);

    }

    // we have established that this is a covariant upcast... though

    // coders are invariant, we are just checking one direction

    @SuppressWarnings("unchecked")

	TypeDescriptor<T> candidateOkDescriptor = (TypeDescriptor<T>) candidateDescriptor;



    // If the coded type is a parameterized type where any of the actual

    // type parameters are not compatible, then the whole thing is certainly not

    // compatible.

    if ((codedType instanceof ParameterizedType) && !isNullOrEmpty(coder.getCoderArguments())) {

      ParameterizedType parameterizedSupertype =

          (ParameterizedType) candidateOkDescriptor.getSupertype(codedClass).getType();

      Type[] typeArguments = parameterizedSupertype.getActualTypeArguments();

      List<? extends Coder<?>> typeArgumentCoders = coder.getCoderArguments();

      if (typeArguments.length < typeArgumentCoders.size()) {

        throw new IncompatibleCoderException(

            String.format(

                "Cannot encode elements of type %s with coder %s:"

                    + " the generic supertype %s has %s type parameters, which is less than the"

                    + " number of coder arguments %s has (%s).",

                candidateOkDescriptor,

                coder,

                parameterizedSupertype,

                typeArguments.length,

                coder,

                typeArgumentCoders.size()),

            coder,

            candidateOkDescriptor.getType());

      }

      for (int i = 0; i < typeArgumentCoders.size(); i++) {

        try {

          Coder<?> typeArgumentCoder = typeArgumentCoders.get(i);

          verifyCompatible(

              typeArgumentCoder, candidateDescriptor.resolveType(typeArguments[i]).getType());

        } catch (IncompatibleCoderException exn) {

          throw new IncompatibleCoderException(

              String.format(

                  "Cannot encode elements of type %s with coder %s"

                      + " because some component coder is incompatible",

                  candidateType, coder),

              coder,

              candidateType,

              exn);

        }

      }

    }

  }



  private static boolean isNullOrEmpty(Collection<?> c) {

    return c == null || c.isEmpty();

  }



  /** The list of {@link CoderProvider coder providers} to use to provide Coders. */

  private ArrayDeque<CoderProvider> coderProviders;



  /**

   * Returns a {@link Coder} to use for values of the given type, in a context where the given types

   * use the given coders.

   *

   * @throws CannotProvideCoderException if a coder cannot be provided

   */

  private <T> Coder<T> getCoderFromTypeDescriptor(

	  TypeDescriptor<T> typeDescriptor, SetMultimap<Type, Coder<?>> typeCoderBindings)

      throws CannotProvideCoderException {

    Type type = typeDescriptor.getType();

    Coder<?> coder;

    if (typeCoderBindings.containsKey(type)) {

      Set<Coder<?>> coders = typeCoderBindings.get(type);

      if (coders.size() == 1) {

        coder = Iterables.getOnlyElement(coders);

      } else {

        throw new CannotProvideCoderException(

            String.format(

                "Cannot provide a coder for type variable %s"

                    + " because the actual type is over specified by multiple"

                    + " incompatible coders %s.",

                type, coders),

            CannotProvideCoderException.ReasonCode.OVER_SPECIFIED);

      }

    } else if (type instanceof Class<?>) {

      coder = getCoderFromFactories(typeDescriptor, Collections.emptyList());

    } else if (type instanceof ParameterizedType) {

      coder = getCoderFromParameterizedType((ParameterizedType) type, typeCoderBindings);

    } else if (type instanceof TypeVariable) {

      coder = getCoderFromFactories(typeDescriptor, Collections.emptyList());

    } else if (type instanceof WildcardType) {

      // No coder for an unknown generic type.

      throw new CannotProvideCoderException(

          String.format("Cannot provide a coder for wildcard type %s.", type), CannotProvideCoderException.ReasonCode.UNKNOWN);

    } else {

      throw new RuntimeException("Internal error: unexpected kind of Type: " + type);

    }



    LOG.debug("Coder for {}: {}", typeDescriptor, coder);

    @SuppressWarnings("unchecked")

	Coder<T> result = (Coder<T>) coder;

    return result;

  }



  /**

   * Returns a {@link Coder} to use for values of the given parameterized type, in a context where

   * the given types use the given {@link Coder Coders}.

   *

   * @throws CannotProvideCoderException if no coder can be provided

   */

  private Coder<?> getCoderFromParameterizedType(

      ParameterizedType type, SetMultimap<Type, Coder<?>> typeCoderBindings)

      throws CannotProvideCoderException {



    List<Coder<?>> typeArgumentCoders = new ArrayList<>();

    for (Type typeArgument : type.getActualTypeArguments()) {

      try {

        Coder<?> typeArgumentCoder =

            getCoderFromTypeDescriptor(TypeDescriptor.of(typeArgument), typeCoderBindings);

        typeArgumentCoders.add(typeArgumentCoder);

      } catch (CannotProvideCoderException exc) {

        throw new CannotProvideCoderException(

            String.format(

                "Cannot provide coder for parameterized type %s: %s", type, exc.getMessage()),

            exc);

      }

    }

    return getCoderFromFactories(TypeDescriptor.of(type), typeArgumentCoders);

  }



  /**

   * Attempts to create a {@link Coder} from any registered {@link CoderProvider} returning the

   * first successfully created instance.

   */

  private Coder<?> getCoderFromFactories(

	  TypeDescriptor<?> typeDescriptor, List<Coder<?>> typeArgumentCoders)

      throws CannotProvideCoderException {

    List<CannotProvideCoderException> suppressedExceptions = new ArrayList<>();

    for (CoderProvider coderProvider : coderProviders) {

      try {

        return coderProvider.coderFor(typeDescriptor, typeArgumentCoders);

      } catch (CannotProvideCoderException e) {

        // Add all failures as suppressed exceptions.

        suppressedExceptions.add(e);

      }

    }



    // Build up the error message and list of causes.

    StringBuilder messageBuilder =

        new StringBuilder()

            .append("Unable to provide a Coder for ")

            .append(typeDescriptor)

            .append(".\n")

            .append("  Building a Coder using a registered CoderProvider failed.\n")

            .append("  See suppressed exceptions for detailed failures.");

    CannotProvideCoderException exceptionOnFailure =

        new CannotProvideCoderException(messageBuilder.toString());

    for (CannotProvideCoderException suppressedException : suppressedExceptions) {

      exceptionOnFailure.addSuppressed(suppressedException);

    }

    throw exceptionOnFailure;

  }



  /**

   * Returns an immutable {@code SetMultimap} from each of the type variables embedded in the given

   * type to the corresponding types in the given {@link Coder}.

   */

  private SetMultimap<Type, Coder<?>> getTypeToCoderBindings(Type type, Coder<?> coder) {

    checkArgument(type != null);

    checkArgument(coder != null);

    if (type instanceof TypeVariable || type instanceof Class) {

      return ImmutableSetMultimap.of(type, coder);

    } else if (type instanceof ParameterizedType) {

      return getTypeToCoderBindings((ParameterizedType) type, coder);

    } else {

      return ImmutableSetMultimap.of();

    }

  }



  /**

   * Returns an immutable {@code SetMultimap} from the type arguments of the parameterized type to

   * their corresponding {@link Coder Coders}, and so on recursively for their type parameters.

   *

   * <p>This method is simply a specialization to break out the most elaborate case of {@link

   * #getTypeToCoderBindings(Type, Coder)}.

   */

  private SetMultimap<Type, Coder<?>> getTypeToCoderBindings(

      ParameterizedType type, Coder<?> coder) {

    List<Type> typeArguments = Arrays.asList(type.getActualTypeArguments());

    List<? extends Coder<?>> coderArguments = coder.getCoderArguments();



    if ((coderArguments == null) || (typeArguments.size() != coderArguments.size())) {

      return ImmutableSetMultimap.of();

    } else {

      SetMultimap<Type, Coder<?>> typeToCoder = HashMultimap.create();



      typeToCoder.put(type, coder);



      for (int i = 0; i < typeArguments.size(); i++) {

        Type typeArgument = typeArguments.get(i);

        Coder<?> coderArgument = coderArguments.get(i);

        if (coderArgument != null) {

          typeToCoder.putAll(getTypeToCoderBindings(typeArgument, coderArgument));

        }

      }



      return ImmutableSetMultimap.<Type, Coder<?>>builder().putAll(typeToCoder).build();

    }

  }

}